package com.jorado.elasticsearch;

import com.jorado.core.utility.StringUtils;
import com.jorado.elasticsearch.model.ESCluster;
import com.jorado.event.EventClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;

public class ElasticSearchClientFactory {

    private final static Integer DEFAULT_POOL_SIZE = 10;
    private static Map<Integer, List<TransportClient>> clientPools = new ConcurrentHashMap<>();

    /**
     * 创建ES连接
     *
     * @param usePools  是否使用连接池（不使用连接池每次创建新连接，使用完后将关闭连接【每个连接只能使用一次】）
     * @param esCluster ES集群配置信息
     * @return
     */
    public static ElasticSearchClient createESClient(boolean usePools, ESCluster esCluster) {
        return createESClient(usePools, DEFAULT_POOL_SIZE, esCluster);
    }

    /**
     * 创建ES连接（使用连接池，连接池大小默认为10）
     *
     * @param esCluster ES集群配置信息
     * @return
     */
    public static ElasticSearchClient createESClient(ESCluster esCluster) {
        return createESClient(true, DEFAULT_POOL_SIZE, esCluster);
    }

    /**
     * 创建ES连接
     *
     * @param usePools  是否使用连接池（不使用连接池每次创建新连接，使用完后将关闭连接【每个连接只能使用一次】）
     * @param poolSize  连接池大小（默认为10）
     * @param esCluster ES集群配置信息
     * @return
     */
    public static ElasticSearchClient createESClient(boolean usePools, int poolSize, ESCluster esCluster) {
        if (esCluster == null || esCluster.getHosts() == null || esCluster.getHosts().size() == 0) {
            EventClient.getDefault().createLog("ESCluster配置错误！").asyncSubmit();
            return null;
        }

        TransportClient client = null;

        if (usePools) {
            Integer key = esCluster.hashCode();
            int index = ThreadLocalRandom.current().nextInt(poolSize);
            if (clientPools.containsKey(key) && clientPools.get(key).size() == poolSize) {
                client = clientPools.get(key).get(index);
                if (clientIsNullOrClosed(client))
                    synchronized (clientPools) {
                        if (clientIsNullOrClosed(client))
                            clientPools.get(key).remove(index);
                    }
            }
            if (clientIsNullOrClosed(client)) {
                synchronized (clientPools) {
                    if (clientIsNullOrClosed(client)) {
                        if (!clientPools.containsKey(key) || clientPools.get(key) == null || clientPools.get(key).size() == 0 || clientPools.get(key).size() != poolSize) {
                            if (clientPools.containsKey(key) && clientPools.get(key).size() != poolSize) {
                                EventClient.getDefault().createException("连接池不足，重新创建").asyncSubmit();
                                clientPools.remove(key);
                            }

                            List<TransportClient> clients = new ArrayList<>();
                            for (int i = 0; i < poolSize; i++) {
                                TransportClient client1 = createESClient(esCluster.getHosts(), esCluster.getDefaultPort(), esCluster.getSettings());
                                clients.add(client1);
                            }
                            clientPools.put(key, clients);
                            client = clients.get(index);
                            EventClient.getDefault().createLog("构建连接池完成").asyncSubmit();
                        } else {
                            client = createESClient(esCluster.getHosts(), esCluster.getDefaultPort(), esCluster.getSettings());
                            clientPools.get(key).set(index, client);
                            EventClient.getDefault().createLog("连接池中获取的连接为NULL，重新创建并填充").asyncSubmit();
                        }
                    }
                }
            }
        } else
            client = createESClient(esCluster.getHosts(), esCluster.getDefaultPort(), esCluster.getSettings());

        return new ElasticSearchClient(!usePools, client);
    }

    private static boolean clientIsNullOrClosed(TransportClient client) {
        return client == null || client.connectedNodes() == null || client.connectedNodes().size() == 0;
    }

    /**
     * 创建ES连接
     *
     * @param esHosts     Host List
     * @param defaultPort Es port
     * @param setting     Settings
     * @return
     * @throws UnknownHostException
     */
    private static TransportClient createESClient(List<String> esHosts, int defaultPort, Map<String, Object> setting) {
        if (esHosts == null || esHosts.size() == 0)
            return null;

        Settings.Builder builder = Settings.builder();
        if (setting == null || setting.size() == 0) {
            setting = new HashMap<>();
            setting.put("client.transport.ignore_cluster_name", "true");
        }

        Settings settings = builder.put(setting).build();

        List<TransportAddress> transportAddresses = new ArrayList<>();
        for (String esHost : esHosts) {
            String[] hostPort = esHost.split(":");
            String host;
            if (!StringUtils.isNullOrWhiteSpace(hostPort[0])) {
                host = hostPort[0];
                int port = hostPort.length == 2 && !StringUtils.isNullOrWhiteSpace(hostPort[1]) ? Integer.valueOf(hostPort[1]) : defaultPort;
                try {
                    transportAddresses.add(new InetSocketTransportAddress(InetAddress.getByName(host), port));
                } catch (UnknownHostException e) {
                    EventClient.getDefault().createException("构建ES Client").setException(e).asyncSubmit();
                }
            }
        }
        return new PreBuiltTransportClient(settings).addTransportAddresses(transportAddresses.toArray(new TransportAddress[transportAddresses.size()]));
    }
}
